package com.vincent.acceptor;

import com.vincent.common.Environment;
import com.vincent.common.Message;
import com.vincent.common.Process;
import com.vincent.message.P1aMessage;
import com.vincent.message.P1bMessage;
import com.vincent.message.P2aMessage;
import com.vincent.message.P2bMessage;
import com.vincent.utils.BallotNumber;
import com.vincent.utils.PValue;
import com.vincent.utils.PValueSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Acceptor主要维持Paxos的容错内存和p1a、p2a消息的回复，Acceptor维持两个变量：
 * 1、ballot_number：
 * 2、accepted：
 * Created by wensen on 2016/12/12.
 */
public class Acceptor extends Process{

    private static Logger LOGGER = LoggerFactory.getLogger(Acceptor.class);

    private BallotNumber ballot_number;
    private PValueSet accepted;

    public Acceptor(Environment env, int id) {
        super(env, id);
        this.ballot_number = null;
        this.accepted = new PValueSet();
    }

    /**
     *Acceptor接收p1a和p2a消息：
     *
     * p1a消息：一个leader的请求消息，对应相应的ballot number
     * acceptor作以下的状态转移：首先如果该leader的ballot number大于自身当前的
     * ballot number，则采用(adopt),并回复一个p1b消息包含当前的ballot number以及
     * 已接受(accept)的值
     *
     * p2a消息：从一个leader发送过来的，包含了pvalue(b, s, c)，acceptor将作以下转移：
     * 如果当前的ballot number等于b，那么acceptor僵接受(b,s,c),并给leader回复p2b消息
     * 其中包含它当前的ballot number
     */
    public void body() {

        LOGGER.debug("Acceptor: Here I am: {}", this.id());

        while (true) {

            Message msg = null;
            try {
                msg = this.getNextMessage();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            if (msg instanceof P1aMessage) {

                if (((P1aMessage) msg).ballot_number().compareTo(this.ballot_number) > 0) {
                    this.ballot_number = ((P1aMessage) msg).ballot_number();
                }
                this.sendMessage(msg.src(), new P1bMessage(this.id(), this.ballot_number, this.accepted));
            } else if (msg instanceof P2aMessage) {

                if (((P2aMessage) msg).ballot_number().compareTo(this.ballot_number) == 0) {
                    this.accepted.add(new PValue(((P2aMessage) msg).ballot_number(), ((P2aMessage) msg).slot_number(), ((P2aMessage) msg).command()));
                }
                this.sendMessage(msg.src(), new P2bMessage(this.id(), this.ballot_number, ((P2aMessage) msg).slot_number()));
            }

        }

    }
}
